手写dubbo 10-基于netty实现RPC | 并发笔记

博客中代码地址：https://github.com/farliu/farpc.git
dubbo架构

本文实现的是远程调用，也就是图片中的第4步，dubbo作为一款RPC框架，这是它的核心功能，dubbo提供了很多种方式，如下图：
dubbo module

原理分析

首先科普一下RPC三个字母，即Remote Procedure Call。简单来说就是从一台机器（客户端）上通过参数传递的方式调用另一台机器（服务器）上的一个函数或方法（可以统称为服务）并得到返回的结果。

回想一下，java本地方法调用。假如在电脑ComputerA上有一个类ClassA，其中有methodA()方法，我们调用的话，就是new一个ClassA的对象classA，然后classA.methodA()来调用。这其中值得深思，凭什么我们能new一个ClassA的对象？我们能不能new一个ComputerB机器上的类ClassB的对象呢？

第二个答案肯定都知道，肯定是不行。第一个问题，我们为什么能在ComputerA上new一个ClassA的对象，是因为在ComputerA上存在ClassA类的class文件，通过JVM加载后，我们可以实现对它的调用。所以为了解决第二个问题，前辈们还真想出了办法。你不是只要class文件就行了嘛，那就给你，ComputerA打个jar包给ComputerB去加载。也就达到了调用的效果。

以上其实就是解决两个应用之间交互的早期办法。但是当系统依赖复杂后，这种方式极为不妥，每一个系统都得加载子系统的所有class，非常不合适。以及还有敏感代码允许泄露等问题。

为了解决这个问题，我们也就出现RPC服务，就是当ComputerA要调用ComputerB的方法时，ComputerA通过某种方式告诉ComputerB，再由ComputerB执行完之后，将结果告诉ComputerA。这就是RPC最初的设想。我们归纳一下几个步骤

ComputerA将自己的需要调用的方法和参数准备封装好。
按照约定的方式，将封装好的参数传给ComputerB
ComputerB收到约定的数据后，解析获得ComputerA需要调用的方法和参数。
ComputerB按照ComputerA给的数据，执行对应的方法。
ComputerB将执行结果按照约定返回ComputerA。

可以看到这个过程，极为重要的就是数据传输，为了实现数据传输，我们搞出了很多花样。比较通常的就是，将参数转成xml，通过Http传送给另一台机器，后来发现xml体积太大，我们又将json代替了xml。再后来我们又觉得每次用http协议，都得重新连接，又使用socket实现长连接。再后来觉得socket实现阻塞IO，效率不高，又推送了NIO，以及selector、channel这些专业术语。这都是在优化传输过程，本章采用netty来实现传输。

对于ComputerB执行相应的方法，基于以上的约定，ComputerB拿到所需的参数后，使用java反射就能调用具体的方法了。

项目结构介绍

项目结构
本节涉及博客中代码的module，farpc-rpc（远程调用）、farpc-demo。

初始化netty

本章使用netty实现rpc，自然要导入jar包。

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.36.Final</version>
</dependency>

秉承可扩展设计，提供两个接口。

@FarSPI("netty")
public interface IProviderServer {
    void start(String selfAddress);
}

@FarSPI("netty")
public interface IConsumerServer {
    Object execute(String address, RequestDTO requestDTO);
}

服务端

服务端就是常规的netty代码，启动服务，然后配置Handler，处理接收的信息。值得一提的是，在服务端启动的时候，我会去扫描所有标注了Provider注解的类，然后将其注入到注册中心，并为一个container保存对应的对象，用于反射执行指定方法。

Provider模拟dubbo中的@Service注解，用于注册服务。

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE})
public @interface Provider {
    Class interfaceClazz();

    String name() default "";
}

Container扫描被Provider修饰的类，然后反射生成对象，将其保存到本地容器供反射执行指定的方法、保存到注册中心供消费端发现服务。

public class Container {
    private static final Logger logger = LoggerFactory.getLogger(Container.class);
    private static IRegistrar registrar = RegistrarFactory.getRegistrar();
    private static Map<String, Object> providers = new HashMap<String, Object>();

    static {
        Reflections reflections = new Reflections(new ConfigurationBuilder()
                .setUrls(ClasspathHelper.forPackage("com.ofcoder"))
                .setScanners(new TypeAnnotationsScanner()));
        Set<Class<?>> classes = reflections.getTypesAnnotatedWith(Provider.class, true);
        for (Class<?> clazz : classes) {
            try {
                Provider annotation = clazz.getAnnotation(Provider.class);
                Object provider = clazz.newInstance();
                String canonicalName = annotation.interfaceClazz().getCanonicalName();

                // 保存到本地容器
                providers.put(canonicalName, provider);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    public static void registerSelf(String selfAddress){
        for (String service : providers.keySet()) {
            registrar.register(selfAddress, service);
        }
    }

    public static Map<String, Object> getProviders() {
        return providers;
    }
}

NettyProviderHandler用来处理收到的信息，然后根据收到的数据，从本地容器中取得对象，调用指定的方法，并将执行结果返回给消费端。代码如下：

public class NettyProviderHandler extends ChannelInboundHandlerAdapter {
    private static final Logger logger = LoggerFactory.getLogger(NettyProviderHandler.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
        RequestDTO requestDTO = (RequestDTO) msg;
        Object result = new Object();

        logger.info("receive request.. {}", requestDTO);
        if (Container.getProviders().containsKey(requestDTO.getClassName())) {
            Object provider = Container.getProviders().get(requestDTO.getClassName());

            Class<?> providerClazz = provider.getClass();
            Method method = providerClazz.getMethod(requestDTO.getMethodName(), requestDTO.getTypes());
            // 反射执行指定的方法
            result = method.invoke(provider, requestDTO.getParams());
        }

        // 将结果输出到消费端
        ctx.write(result);
        ctx.flush();
        ctx.close();
    }
}

NettyProviderServer就是监听指定的端口，启动服务。

public class NettyProviderServer implements IProviderServer {
    private static final Logger logger = LoggerFactory.getLogger(NettyProviderServer.class);

    public void start(String selfAddress) {
        Container.registerSelf(selfAddress);

        String[] addrs = selfAddress.split(":");
        String ip = addrs[0];
        Integer port = Integer.parseInt(addrs[1]);

        publisher(ip, port);
    }

    private void publisher(String ip, Integer port) {
        // 启动服务
        try {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();

            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            pipeline.addLast(new ObjectEncoder());
                            pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(NettyProviderServer.class.getClassLoader())));
                            pipeline.addLast(new NettyProviderHandler());
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture future = bootstrap.bind(ip, port).sync();
            logger.info("netty server is started...");
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}

消费端

NettyConsumerServer用于发送请求，将封装好的参数发给服务提供者。

public class NettyConsumerServer implements IConsumerServer {
    private static final Logger logger = LoggerFactory.getLogger(NettyConsumerServer.class);

    public Object execute(String serivceAddress, RequestDTO requestDTO) {
        String[] addrs = serivceAddress.split(":");
        String host = addrs[0];
        Integer port = Integer.parseInt(addrs[1]);

        final NettyConsumerHandler consumerHandler = new NettyConsumerHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            pipeline.addLast( new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(ConsumerProxy.class.getClassLoader())));
                            pipeline.addLast( new ObjectEncoder());
                            pipeline.addLast(consumerHandler);
                        }
                    });
            ChannelFuture future = bootstrap.connect(host, port).sync();

            Channel channel = future.channel();
            channel.writeAndFlush(requestDTO);
            logger.info("send request..., {}", requestDTO);
            channel.closeFuture().sync();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            group.shutdownGracefully();
        }
        return consumerHandler.getResponse();

    }
}

NettyConsumerHandler用于处理提供端返回的结果，这里没有做过多处理，直接返回。

public class NettyConsumerHandler extends ChannelInboundHandlerAdapter {
    private Object response;

    public Object getResponse() {
        return response;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        response = msg;
    }
}

使用SPI整合

在上一章，已经把SPI集成的很不错了，这里我们可以按照上一章套路管理RPC服务。我们提供一个Factory，用来替代自适应扩展。

public class RpcFactory {
    public static IConsumerServer getConsumerService() {
        String protocol = Property.Rpc.protocol;
        IConsumerServer extension = ExtensionLoader.getExtensionLoader(IConsumerServer.class).getExtension(protocol);
        return extension;
    }

    public static IProviderServer getProviderServer() {
        String protocol = Property.Rpc.protocol;
        IProviderServer extension = ExtensionLoader.getExtensionLoader(IProviderServer.class).getExtension(protocol);
        return extension;
    }
}

配置文件

com.ofcoder.farpc.rpc.IConsumerServer
netty=com.ofcoder.farpc.rpc.netty.NettyConsumerServer

------------------------------------------------------

com.ofcoder.farpc.rpc.IProviderServer
netty=com.ofcoder.farpc.rpc.netty.NettyProviderServer

测试

@Test
public void provider() throws IOException {
    IProviderServer providerServer = RpcFactory.getProviderServer();
    providerServer.start("127.0.0.1:20880");
    System.in.read();
}

@Test
public void consumer() {
    IConsumerServer consumerService = RpcFactory.getConsumerService();
    Object execute = consumerService.execute("127.0.0.1:20880", new RequestDTO());
}

先启动提供者，然后在执行消费。可以在提供者的控制台看到，相应的日志，也就说明达到我们需要的效果了。如下

1
2
3

...
main  INFO netty.NettyProviderServer: netty server is started...
nioEventLoopGroup-3-1  INFO netty.NettyProviderHandler: receive request.. RequestDTO{className='null', methodName='null', types=null, params=null}

dubbo源码

这一节，来印证dubbo实现过程，也算是当作源码导读。这一节也分为两段分析，一消费端调用过程，二服务端收到请求处理过程。

整个过程可以总结为：首先服务消费者通过代理对象 Proxy 发起远程调用，接着通过网络客户端 Client 将编码后的请求发送给服务提供方的网络层上，也就是 Server。Server 在收到请求后，首先要做的事情是对数据包进行解码。然后将解码后的请求发送至分发器 Dispatcher，再由分发器将请求派发到指定的线程池上，最后由线程池调用具体的服务。这就是一个远程调用请求的发送与接收过程。

消费端

消费端调用复杂在于链路太长，由Proxy调用开始，会经过一系列的Invoker，直到DubboInvoker再去真正发起请求

Proxy.greet
  -> InvokerInvocationHandler.invoke
    -> MockClusterInvoker.invoke
      -> ...
        -> AbstractInvoker.invoke
          -> DubboInvoker.doInvoke

InvokerInvocationHandler用于排除调用toString()、equals()、hashCode()这些方法。MockClusterInvoker主要实现了降级逻辑，在服务调用失败后用于返回默认值。以及后续还有FailoverClusterInvoker，最后会调到DubboInvoker.doInvoke，着重关心这一块，中间那些增强的Invoker逻辑，可以自己了解。

protected Result doInvoke(final Invocation invocation) throws Throwable {
    RpcInvocation inv = (RpcInvocation) invocation;
    final String methodName = RpcUtils.getMethodName(invocation);
    inv.setAttachment(PATH_KEY, getUrl().getPath());
    inv.setAttachment(VERSION_KEY, version);

    ExchangeClient currentClient;
    if (clients.length == 1) {
        currentClient = clients[0];
    } else {
        currentClient = clients[index.getAndIncrement() % clients.length];
    }
    try {
        // 是否有返回值，true表示没有返回值。
        boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
        int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
        if (isOneway) {
            boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
            // 发送请求
            currentClient.send(inv, isSent);
            // 清空future 
            RpcContext.getContext().setFuture(null);
            // 不用关注返回值，返回默认RpcResult
            return AsyncRpcResult.newDefaultAsyncResult(invocation);
        } else {
            //该类实现Future接口，用于实现异步
            AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
            // 发起调用，也返回的Future对象
            CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
            // 当发起调用的Future完成后，会通知到asyncRpcResult
            asyncRpcResult.subscribeTo(responseFuture);
            RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
            return asyncRpcResult;
        }
    } catch (TimeoutException e) {
        ...
    } catch (RemotingException e) {
        ...
    }
}

消费端到此处，应分为第二段，既发送请求。这里dubbo会调用一系列Client，这些Client均是NettyClient的包装增强，对request()的调用都是一直往下传递。

ReferenceCountExchangeClient.request
  -> HeaderExchangeClient.request
    -> HeaderExchangeChannel.request
      -> AbstractClient.send
        -> NettyChannel.send
          -> NioClientSocketChannel.write

ReferenceCountExchangeClient为对象引用增加计数器，当close()调用时，该计数器减1。HeaderExchangeClient增加心跳检测，这里请求会转到Channel对象，并结束request()传递调用。HeaderExchangeChannel对Request的封装，并通过AbstractClient.getChannel()获取到NettyChannel对象并调用其send()方法。完成整个请求的发送。

public void send(Object message, boolean sent) throws RemotingException {
    super.send(message, sent);

    boolean success = true;
    int timeout = 0;
    try {
        // 发送消息(包含请求和响应消息)
        ChannelFuture future = channel.write(message);
        
        // sent 的值源于 <dubbo:method sent="true/false" /> 中 sent 的配置值，有两种配置值：
        //   1. true: 等待消息发出，消息发送失败将抛出异常
        //   2. false: 不等待消息发出，将消息放入 IO 队列，即刻返回
        // 默认情况下 sent = false；
        if (sent) {
            timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            // 等待消息发出，若在规定时间没能发出，success 会被置为 false
            success = future.await(timeout);
        }
        Throwable cause = future.getCause();
        if (cause != null) {
            throw cause;
        }
    } catch (Throwable e) {
        throw new RemotingException(this, "Failed to send message ...");
    }

    // 若 success 为 false，这里抛出异常
    if (!success) {
        throw new RemotingException(this, "Failed to send message ...");
    }
}

提供端

在服务提供端获取到请求之后，然后交由NettyHandler.messageReceived处理，该方法会进行和消费端相关操作，以此会执行MultiMessageHandlerH、eartbeatHandler，最后由ChannelHandler将操作逻辑封装到Runable对象中，交给线程池进行调用处理，这个过程也成为线程派发，dubbo提供5中派发模式。

策略	用途
all	所有消息都派发到线程池，包括请求，响应，连接事件，断开事件等
direct	所有消息都不派发到线程池，全部在 IO 线程上直接执行
message	只有请求和响应消息派发到线程池，其它消息均在 IO 线程上执行
execution	只有请求消息派发到线程池，不含响应。其它消息均在 IO 线程上执行
connection	在 IO 线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到线程池

dubbo默认选择策略是all，整理上述调用链路如下

NettyHandler#messageReceived
  -> AbstractPeer#received
    —> MultiMessageHandler#received
      —> HeartbeatHandler#received
        —> ALLChannelHandler#received
          —> ExecutorService#execute

ALLChannelHandler.received会初始化ChannelEventRunnable对象，由该对象真正完成调用，我们看看具体源码

public class ChannelEventRunnable implements Runnable {
    @Override
    public void run() {
        // 检测通道状态，对于请求或响应消息，此时 state = RECEIVED
        if (state == ChannelState.RECEIVED) {
            try {
                handler.received(channel, message);
            } catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                        + ", message is " + message, e);
            }
        } else {
            switch (state) {
            case CONNECTED:
                ...
                break;
            case DISCONNECTED:
                ...
                break;
            case SENT:
                ...
                break;
            case CAUGHT:
                ...
                break;
            default:
                logger.warn("unknown state: " + state + ", message is " + message);
            }
        }

    }
}

这里多说一句，先用if判断出现频率比较高的消息类型，然后用switch处理其他类型，不用把频率较高的类型和普通类型同级判断，以此提高效率。我们开发过程中也可借鉴这一点。
ChannelEventRunnable作用类似于路由，将消息分别交给各自的ChannelHandler去处理，这里的对象为DecodeHandler，该Handler就是对Request或Response进行解码后，继续传递到HeaderExchangeHandler。

public class HeaderExchangeHandler implements ChannelHandlerDelegate {
    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
        final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
        try {
            if (message instanceof Request) {
                // handle request.
                Request request = (Request) message;
                if (request.isEvent()) {
                    handlerEvent(channel, request);
                } else {
                    // 是否为单/双向调用，判断是否需要接收返回结果
                    if (request.isTwoWay()) {
                        handleRequest(exchangeChannel, request);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } else if (message instanceof Response) {
                handleResponse(channel, (Response) message);
            } else if (message instanceof String) {
                // telnet 相关
                ...
            } else {
                handler.received(exchangeChannel, message);
            }
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }

    Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
        Response res = new Response(req.getId(), req.getVersion());
        // 检测请求是否合法，不合法则返回状态码为 BAD_REQUEST 的响应
        if (req.isBroken()) {
            Object data = req.getData();

            String msg;
            if (data == null)
                msg = null;
            else if
                (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
            else
                msg = data.toString();
            res.setErrorMessage("Fail to decode request due to: " + msg);
            // 设置 BAD_REQUEST 状态
            res.setStatus(Response.BAD_REQUEST);

            return res;
        }
        
        // 获取 data 字段值，也就是 RpcInvocation 对象
        Object msg = req.getData();
        try {
            // 继续向下调用
            Object result = handler.reply(channel, msg);
            // 设置 OK 状态码
            res.setStatus(Response.OK);
            // 设置调用结果
            res.setResult(result);
        } catch (Throwable e) {
            // 若调用过程出现异常，则设置 SERVICE_ERROR，表示服务端异常
            res.setStatus(Response.SERVICE_ERROR);
            res.setErrorMessage(StringUtils.toString(e));
        }
        return res;
    }
}

接下来要说的就是定义在DubboProtocol类中匿名对象的reply方法，既ExchangeHandlerAdapter.reply()

public class DubboProtocol extends AbstractProtocol {
    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

        @Override
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                // 获取 Invoker 实例
                Invoker<?> invoker = getInvoker(channel, inv);
                if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                    // 回调相关，忽略
                }
                RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                // 通过 Invoker 调用具体的服务
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: ...");
        }
        ...
    }

先通过getInvoker()获取Invoker实例，然后调用invoke方法。getInvoker()方法先从缓存中获取，没命中则调用DubboExporter.getInvoker()继续创建。而Invoker的invoke方法是由AbstractProxyInvoker实现，

public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
    @Override
    public Result invoke(Invocation invocation) throws RpcException {
        try {
            Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());

            // 将结果封装到AsyncRpcResult，然后返回
            ...
            return asyncRpcResult;
        } catch (InvocationTargetException e) {
            ...
            return AsyncRpcResult.newDefaultAsyncResult(null, e.getTargetException(), invocation);
        } catch (Throwable e) {
            throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }
}

剩余的最后一个doInvoke，是一个抽象方法，由子类实现，而Invoker实现类是由JavassistProxyFactory 动态生成，具体可查看JavassistProxyFactory.getInvoker()方法。最后生成的代理类逻辑如下：

/** Wrapper0 是在运行时生成的，可使用 Arthas 进行反编译 */
public class Wrapper0 extends Wrapper implements ClassGenerator.DC {
    public static String[] pns;
    public static Map pts;
    public static String[] mns;
    public static String[] dmns;
    public static Class[] mts0;

    // 省略其他方法

    public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException {
        DemoService demoService;
        try {
            // 类型转换
            demoService = (DemoService)object;
        }
        catch (Throwable throwable) {
            throw new IllegalArgumentException(throwable);
        }
        try {
            // 根据方法名调用指定的方法
            if ("sayHello".equals(string) && arrclass.length == 1) {
                return demoService.sayHello((String)arrobject[0]);
            }
        }
        catch (Throwable throwable) {
            throw new InvocationTargetException(throwable);
        }
        throw new NoSuchMethodException(new StringBuffer().append("Not found method \"").append(string).append("\" in class com.alibaba.dubbo.demo.DemoService.").toString());
    }
}

这里可以看到，最后不是通过反射去执行的，而是根据具体方法名路由的，然后调用执行的。所以以后谁要说反射执行，就拿这篇文章呼他的脸。不过话说回来，我们实现的调用还是通过反射，我们也看到了dubbo的实现太复杂了，如果再造一个轮子没必要，主要是弄清楚原理。